package com.cfh.springcloudserver.listener;

import com.cfh.springcloudserver.stream.SimpleMessageSink;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.UnsupportedEncodingException;

@Component
public class SimpleMessageListener {

    @Autowired(required = false)
    private SimpleMessageSink simpleMessageSink;


    @PostConstruct
    public void init() {
        SubscribableChannel subscribableChannel = simpleMessageSink.input();
        subscribableChannel.subscribe(message -> {
            MessageHeaders headers = message.getHeaders();
            String encoding = (String) headers.get("charset-encoding");
            byte[] content = (byte[]) message.getPayload();
            try {
                System.out.println("接收到消息:" + new String(content, encoding));
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        });
    }

    @StreamListener("gupao2018")
    public void onMessage(byte[] content) {
        System.out.println("onMessage(byte[]):" + new String(content));
    }

    @StreamListener("test007")
    public void onMessageRocketmq(byte[] data) {
        System.out.println("ROCKETMQ:" + new String(data));
    }

    @StreamListener("gupao2018")
    public void onMessage(String content) {
        System.out.println("onMessage(String):" + content);
    }

    @ServiceActivator(inputChannel = "gupao2018")
    public void onServiceActivator(String data) {
        System.out.println("onServiceActivator(String):" + data);
    }
}
